import java.util.*;
import java.util.concurrent.*;
import java.time.LocalDateTime;
import java.time.Instant;
import java.time.ZoneId;

/**
 * 简化版IoT数据处理系统
 * 不依赖外部库，可以直接运行演示
 */
public class SimpleApplication {
    
    // 消息队列
    private static final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>(1000);
    
    // 存储解析后的数据（模拟Redis）
    private static final Map<String, String> dataStore = new ConcurrentHashMap<>();
    
    // 统计信息
    private static volatile long producedCount = 0;
    private static volatile long processedCount = 0;
    private static volatile long successCount = 0;
    private static volatile long errorCount = 0;
    
    public static void main(String[] args) {
        System.out.println("=== 简化版IoT数据处理系统启动 ===");
        
        // 启动生产者
        Thread producer = new Thread(new DataProducer());
        producer.start();
        
        // 启动消费者
        Thread consumer = new Thread(new DataConsumer());
        consumer.start();
        
        // 监控线程
        Thread monitor = new Thread(new SystemMonitor());
        monitor.start();
        
        // 用户交互
        Scanner scanner = new Scanner(System.in);
        System.out.println("输入 'status' 查看状态, 'stop' 停止系统:");
        
        while (true) {
            String command = scanner.nextLine().trim();
            if ("status".equals(command)) {
                printStatus();
            } else if ("stop".equals(command)) {
                System.out.println("正在停止系统...");
                producer.interrupt();
                consumer.interrupt();
                monitor.interrupt();
                break;
            } else if ("help".equals(command)) {
                System.out.println("可用命令: status, stop, help");
            }
        }
        
        scanner.close();
        System.out.println("系统已停止");
    }
    
    /**
     * 数据生产者
     */
    static class DataProducer implements Runnable {
        @Override
        public void run() {
            Random random = new Random();
            
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // 生成设备ID (1-100)
                    int deviceId = random.nextInt(100) + 1;
                    String deviceIdHex = String.format("%08X", deviceId);
                    
                    // 生成时间戳
                    long timestamp = System.currentTimeMillis() / 1000;
                    String timestampHex = String.format("%08X", (int)timestamp);
                    
                    // 生成温度 (10-40度)
                    double temperature = 10 + random.nextDouble() * 30;
                    int tempInt = (int)(temperature * 100);
                    String tempHex = String.format("%04X", tempInt);
                    
                    // 生成湿度 (30-90%)
                    double humidity = 30 + random.nextDouble() * 60;
                    int humidityInt = (int)(humidity * 100);
                    String humidityHex = String.format("%04X", humidityInt);
                    
                    // 组合消息
                    String message = deviceIdHex + timestampHex + tempHex + humidityHex;
                    
                    // 放入队列
                    if (messageQueue.offer(message)) {
                        producedCount++;
                    }
                    
                    // 控制生产速率 (每秒10条)
                    Thread.sleep(100);
                    
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    System.err.println("生产者异常: " + e.getMessage());
                }
            }
        }
    }
    
    /**
     * 数据消费者
     */
    static class DataConsumer implements Runnable {
        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    String message = messageQueue.poll(1, TimeUnit.SECONDS);
                    if (message != null) {
                        processMessage(message);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    System.err.println("消费者异常: " + e.getMessage());
                    errorCount++;
                }
            }
        }
        
        private void processMessage(String message) {
            processedCount++;
            
            try {
                // 验证消息长度
                if (message.length() != 24) {
                    throw new IllegalArgumentException("消息长度不正确: " + message.length());
                }
                
                // 解析消息
                String deviceIdHex = message.substring(0, 8);
                String timestampHex = message.substring(8, 16);
                String tempHex = message.substring(16, 20);
                String humidityHex = message.substring(20, 24);
                
                // 转换数值
                long deviceId = Long.parseUnsignedLong(deviceIdHex, 16);
                long timestamp = Long.parseUnsignedLong(timestampHex, 16);
                int tempInt = Integer.parseUnsignedInt(tempHex, 16);
                int humidityInt = Integer.parseUnsignedInt(humidityHex, 16);
                
                double temperature = tempInt / 100.0;
                double humidity = humidityInt / 100.0;
                
                // 验证数据范围
                if (temperature < -50 || temperature > 80) {
                    throw new IllegalArgumentException("温度超出范围: " + temperature);
                }
                if (humidity < 0 || humidity > 100) {
                    throw new IllegalArgumentException("湿度超出范围: " + humidity);
                }
                
                // 创建数据对象
                LocalDateTime dateTime = LocalDateTime.ofInstant(
                    Instant.ofEpochSecond(timestamp), ZoneId.systemDefault());
                
                String deviceName = String.format("DEVICE_%08d", deviceId);
                String key = deviceName + ":" + dateTime.toString();
                
                String jsonData = String.format(
                    "{\"deviceId\":\"%s\",\"timestamp\":\"%s\",\"temperature\":%.2f,\"humidity\":%.2f,\"rawMessage\":\"%s\"}",
                    deviceName, dateTime, temperature, humidity, message
                );
                
                // 存储数据（模拟Redis）
                dataStore.put(key, jsonData);
                successCount++;
                
            } catch (Exception e) {
                errorCount++;
                System.err.println("解析消息失败: " + message + " - " + e.getMessage());
            }
        }
    }
    
    /**
     * 系统监控
     */
    static class SystemMonitor implements Runnable {
        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    Thread.sleep(10000); // 每10秒输出一次状态
                    printStatus();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
    
    /**
     * 打印系统状态
     */
    private static void printStatus() {
        double successRate = processedCount > 0 ? (double) successCount / processedCount * 100 : 0;
        
        System.out.println("\n=== 系统状态 ===");
        System.out.println("队列大小: " + messageQueue.size());
        System.out.println("已生产消息: " + producedCount);
        System.out.println("已处理消息: " + processedCount);
        System.out.println("处理成功: " + successCount);
        System.out.println("处理失败: " + errorCount);
        System.out.printf("成功率: %.2f%%\n", successRate);
        System.out.println("存储数据条数: " + dataStore.size());
        
        // 显示最近的几条数据
        if (!dataStore.isEmpty()) {
            System.out.println("\n最近存储的数据示例:");
            dataStore.entrySet().stream()
                .limit(3)
                .forEach(entry -> System.out.println("  " + entry.getKey() + " -> " + entry.getValue()));
        }
        System.out.println();
    }
}